package com.example.cloud.action;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import com.example.cloud.service.SinkSender;

/**
 * kafka
 *
 * @author tongyufu
 */
@RequestMapping("/kafka")
@RestController
public class KafkaAction {

    @Autowired
    private SinkSender     sinkSender;
    @Autowired
    @Qualifier(Sink.INPUT)
    private MessageChannel channel;

    /**
     * 发送消息
     *
     * @param message 消息
     * @return 消息
     */
    @RequestMapping("send/{message}")
    public String send(@PathVariable("message") String message) {
        System.out.println("------send：" + message);
        //channel.send(MessageBuilder.withPayload(message).build());
        sinkSender.input().send(MessageBuilder.withPayload(message).build());
        return message;
    }

}
